package com.db;

import java.io.IOException;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.NamespaceDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.ColumnFamilyDescriptor;
import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.client.TableDescriptor;
import org.apache.hadoop.hbase.client.TableDescriptorBuilder;

/**
 * @author huanggengkeng
 * @description 当前HBase的工具类，主要用于简化操作
 * @date 2021-07-29
 */


public class HBaseUtils {
    final Configuration conf;
    private Admin admin;
    private Connection conn;

    public HBaseUtils(Configuration conf) {
        this.conf = conf;
        init();
    }

    private void init() {
        try {
            getConnection();
            getAdmin();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public HBaseUtils(String zkQuorum,String zkPort) {
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", zkQuorum);
        conf.set("hbase.zookeeper.property.clientPort",zkPort);
        this.conf = conf;
        init();
    }

    public synchronized Connection getConnection() throws IOException {
        if (conn != null) {
            return conn;
        }
        conn =  ConnectionFactory.createConnection(conf);
        return conn;
    }

    public synchronized Admin getAdmin() throws IOException {
        if (admin != null) {
            return admin;
        }
        admin = conn.getAdmin();
        return admin;
    }

    /*** namespace的操作 ***/

    public NamespaceDescriptor[] listNamespace() throws IOException {
        NamespaceDescriptor[] list = getAdmin().listNamespaceDescriptors();
        return list;
    }

    public List<TableDescriptor> listTableDescriptorsByNamespace(String namespace) throws IOException {
        return getAdmin().listTableDescriptorsByNamespace(namespace.getBytes());
    }

    public boolean existsNamespace(String namespace) throws IOException {
        for (NamespaceDescriptor namespaceDescriptor : listNamespace()) {
            if (namespaceDescriptor.getName().equals(namespace)) {
                return true;
            }
        }
        return false;
    }

    public NamespaceDescriptor getNamespace(String namespace) throws IOException {
        if (existsNamespace(namespace)) {
            return getAdmin().getNamespaceDescriptor(namespace);
        }
        return null;
    }

    public NamespaceDescriptor createNamespace(String namespace) throws IOException {
        NamespaceDescriptor namespaceDescriptor = NamespaceDescriptor.create(namespace).build();
        getAdmin().createNamespace(namespaceDescriptor);
        return namespaceDescriptor;
    }

    public boolean deleteNamespace(String namespace) throws IOException {
        getAdmin().deleteNamespace(namespace);
        return true;
    }

    public void modifyNamespaceConfig(NamespaceDescriptor namespaceDescriptor, String key, String value)
            throws IOException {
        namespaceDescriptor.removeConfiguration(key);
        namespaceDescriptor.setConfiguration(key, value);
        getAdmin().modifyNamespace(namespaceDescriptor);
    }

    /*** table的操作 ***/

    public void createTable(String tableName, String... columnFamilies) throws IOException {
        TableName t = TableName.valueOf(tableName);
        TableDescriptorBuilder build = TableDescriptorBuilder.newBuilder(t);
        for (String columnFamily : columnFamilies) {
            ColumnFamilyDescriptor familyDescriptor = ColumnFamilyDescriptorBuilder.of(columnFamily);
            build.setColumnFamily(familyDescriptor);
        }
        TableDescriptor tableDescriptor = build.build();
        getAdmin().createTable(tableDescriptor);
    }

    public TableName[] listTableNames() throws IOException {
        return getAdmin().listTableNames();
    }

    public TableName tableName(String tableName) {
        return TableName.valueOf(tableName);
    }

    public boolean existTable(String tableName) throws IOException {
        return getAdmin().tableExists(tableName(tableName));
    }

    public Table getTable(TableName tableName) throws IOException {
        return conn.getTable(tableName);
    }


    public void disableTable(String tableName) throws IOException {
        getAdmin().disableTable(tableName(tableName));
    }

    public boolean isTableDisabled(String tableName) throws IOException {
        return getAdmin().isTableDisabled(tableName(tableName));
    }

    public boolean deleteTable(String tableName) throws IOException {
        if (!existTable(tableName)) {
            return false;
        }
        if (!isTableDisabled(tableName)) {
            disableTable(tableName);
        }
        getAdmin().deleteTable(TableName.valueOf(tableName));
        return true;
    }

    public void close() throws IOException {
        close(getAdmin());
        close(getConnection());
    }

    private void close(Admin admin) {
        try {
            if (admin != null) {
                admin.close();
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    private void close(Connection conn) {
        try {
            if (conn != null) {
                conn.close();
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}
